package com.lagou.no1

import java.util.Properties

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object DStreamToKafka {
  def main(args: Array[String]): Unit = {
    //设置日志级别
    Logger.getLogger("org").setLevel(Level.ERROR)
    //创建ssc
    val conf: SparkConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName(this.getClass.getCanonicalName)
    val ssc = new StreamingContext(conf, Seconds(5))
    //消费组
    val groupid = "mygroup"
    //创建kafka参数
    val kafkaParam: Map[String, Object] = getKafkaConsumerParameters(groupid)
    //读取数据的主题
    val beforeTopic = Array("sample1")
    //发送到目标主题
    val afterTopic = "sample2"

    //建立流
    val dstream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](beforeTopic, kafkaParam)
    )
    //遍历每个RDD
    dstream.foreachRDD{
      (rdd,_) =>
        //遍历RDD中的每一条数据
        rdd.foreach {
          value =>
            val v: String = value.value() //得到消费到的数据
            //将特殊符号替换为空
            val arr: Array[String] = v.split(",").map{elem => elem.replaceAll("<<<!>>>","")}
            //将数组的每个元素按|拼接
            val str: String = arr.reduce(_ + "|" + _)
            println(str)
            //发送到目标主题
            kafkaProducer(afterTopic,"2",str)


        }
    }

    ssc.start()
    ssc.awaitTermination()


  }
  //获取kafka参数
  def getKafkaConsumerParameters(groupId: String): Map[String, Object] = {
    Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node01:9092,node02:9092,node03:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> groupId,
      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest",
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean)
    )
  }
  //kafka发送数据
  def kafkaProducer(topic:String,key:String,value:String)={
    val brokers = "node01:9092,node02:9092,node03:9092"

    val prop = new Properties()

    //kafka地址
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
    //k,v的序序列化
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])

    val producer = new KafkaProducer[String,String](prop)

    val msg = new ProducerRecord[String, String](topic, key, value)

    producer.send(msg)

    producer.close()
  }
}
